#!/usr/bin/perl

package Commands::Validation;

use parent 'Commands::BaseCommand';
use strict;
use warnings;

use Carp qw(croak);
use JSON;
use POSIX qw(WNOHANG);

use Consts;
use Utils qw(dd json_decode in_array date);
use Log;

our $VERSION = '0.1';

our @actions = (
    {
        method => 'list',
        description => '列出验证相关信息'
    },
    {
        method => 'run',
        description => '执行验证操作'
    }
);

sub new
{
    my $class = shift;
    my $self = $class->SUPER::new();
    return $self;
}

sub list
{
    my $self = shift;
    my @args = @_;
    
    my %validations;
    my $config_path = $Consts::CONFIG_PATH. $Consts::DS. 'validations';
    unless (-d $config_path) {
        Log::error("No validation config found.");
        return;
    }
    if (opendir(my $dir, $config_path)) {
        while (my $file = readdir($dir)) {
            if ($file =~ /\.json$/) {
                my $validation_name = $file;
                $validation_name =~ s/\.json$//;
                $validations{$validation_name} = $self->_getConfigByName($validation_name);
            }
        }
        closedir($dir);
    }

    Log::info("Validation list:");
    for my $validation_name (sort keys %validations) {
        Log::primary("Validation [${validation_name}]");
        my $config = $validations{$validation_name};

        Log::info("   src_connection: $config->{src_connection_name}");
        Log::info("   dst_connection: $config->{dst_connection_name}");
        Log::info("   diff_row_limit: $config->{diff_row_limit}");
        Log::info("   tables:");
        for my $table (@{$config->{tables}}) {
            Log::info("       - $table");
        }
    }
}

sub run
{
    my $self = shift;
    my @args = @_;
    
    my $validation_name = $args[0];
    my $validation_table = $args[1];

    my $start_time = date();
    Log::primary("Start validation at ${start_time}.");

    my %validations;
    if (defined $validation_name) {
        $validations{$validation_name} = $self->_getConfigByName($validation_name);
    } else {
        my $config_path = $Consts::CONFIG_PATH. $Consts::DS. 'validations';
        if (opendir(my $dir, $config_path)) {
            while (my $file = readdir($dir)) {
                if ($file =~ /\.json$/) {
                    $file =~ s/\.json$//;
                    # 验证传参指定的validation
                    if (defined $validation_name && $file ne $validation_name) {
                        Log::info("Skip validation [${file}]");
                        next;
                    }
                    $validations{$file} = $self->_getConfigByName($file);
                }
            }
            closedir($dir);
        }
    }

    my $validation_max_process = _getMaxProcesses($self);
    Log::info("Validation max process: ${validation_max_process}");

    for my $validation_name (sort keys %validations) {
        Log::primary("Start validation [${validation_name}]");
        my $config = $validations{$validation_name};
        Log::info("    Connection, src [ $config->{src_connection_name} ] <-> dst [ $config->{dst_connection_name} ]");

        my @tables = @{ $config->{tables} };
        if (defined $validation_table) {
            unless (in_array($validation_table, @tables)) {
                Log::error("Table [${validation_table}] not found in validation [${validation_name}]");
                return;
            }
            @tables = ($validation_table);
        }
        my $table_count = scalar @tables; 
        my @pids;
    
        for (my $i = 0; $i < $table_count; $i ++) {
            my $table = $tables[$i];

            if (defined $validation_table && $table ne $validation_table) {
                Log::info("    Skip table [${table}]");
                continue;
            }

            # 检查是否达到最大进程数
            while (scalar @pids >= $validation_max_process) {
                my $pid = waitpid(-1, 0);
                if ($pid > 0) {
                    @pids = grep { $_ != $pid } @pids;
                } else {
                    Log::error("Failed to wait for a child process.");
                    last;
                }
            }

            my $pid = fork();
            if ($pid) {
                # 父进程
                push @pids, $pid;
            } elsif ($pid == 0) {
                # 子进程
                Log::info("    Table [${table}] ". ($i + 1) . "/$table_count.");
                
                my $src_db = $self->_getDB($config->{src_connection});
                $src_db->connectDB();
                my $dst_db = $self->_getDB($config->{dst_connection});
                $dst_db->connectDB();

                # 对比行数
                Log::info("    > Start validate table [${table}] row count.");
                my $validate_count_res = $self->_validateCount($src_db, $dst_db, $table);
                if ($validate_count_res) {
                    Log::success("    Table [$table] row count match.");
                } else {
                    Log::warn("    Table [$table] row count mismatch!");
                }

                # 对比数值类型列求和
                Log::info("    > Start validate table [${table}] numeric columns sum.");
                my @src_numeric_columns = $src_db->getNumericColumns($table);
                if (scalar @src_numeric_columns == 0) {
                    Log::success("    No numeric columns found in table [${table}], skip.");
                } else {
                    Log::info("    Found ". scalar @src_numeric_columns. " numeric columns (". join(",", @src_numeric_columns). ") in table [${table}].");
                    for my $column (@src_numeric_columns) {
                        Log::info("    Start validate table [${table}] column [${column}] sum.");
                        my $validate_sum_res = $self->_validateColumnSum($src_db, $dst_db, $table, $column);
                        if ($validate_sum_res) {
                            Log::success("    Table [$table] column [$column] sum match.");
                        } else {
                            Log::warn("    Table [$table] column [$column] sum mismatch!");
                        }
                    }
                }

                # 抽查行对比字符串列
                Log::info("    > Start validate table [${table}] string columns.");
                my @src_string_columns = $src_db->getStringColumns($table);
                if (scalar @src_string_columns == 0) {
                    Log::success("    No string columns found in table [${table}], skip.");
                } else {
                    Log::info("    Found ". scalar @src_string_columns. " string columns (". join(",", @src_string_columns). ") in table [${table}].");

                    my $diff_row_limit = defined $config->{diff_row_limit} ? $config->{diff_row_limit} : 300;
                    my ($allowable, $sampling_conditions, $order) = $src_db->getAmplingConditions($table, @src_string_columns, $diff_row_limit);
                    unless ($allowable) {
                        Log::info("    No sampling conditions found in table [${table}], skip.");
                    } else {
                        Log::info("    Sampling conditions '${sampling_conditions}' in table [${table}].");
                        my $validate_rows_res = $self->_validateRows($src_db, $dst_db, $table, \@src_string_columns, $sampling_conditions, $order, $diff_row_limit);
                        if ($validate_rows_res) {
                            Log::success("    Table [$table] sampling rows match.");
                        } else {
                            Log::warn("    Table [$table] sampling rows mismatch!");
                        }
                    }
                }

                exit 0;
            } else {
                Log::error("Failed to fork a child process.");
            }
        }

        # 等待所有子进程完成
        for my $pid (@pids) {
            waitpid($pid, 0);
        }
    }

    my $finish_time = date();
    Log::primary("Validation done at ${finish_time}.");
}

sub _getMaxProcesses
{
    my $self = shift;
    my $config_ref = $self->getAppConfig(); 
    my %config = %{$config_ref} if ref $config_ref eq 'HASH';
    return $config{validation_max_process};
}

sub _getConfigByName
{
    my ($self, $validation_name) = @_;
    my $config_path = $Consts::CONFIG_PATH. $Consts::DS. 'validations'. $Consts::DS. "${validation_name}.json";
    unless (-f $config_path) {
        Log::error("Validation config [${validation_name}] not found.");
        croak "Validation config [${validation_name}] not found.";
    }
    open(my $fh, '<', $config_path) or croak "Failed to open file: $!";
    my $content = do { local $/; <$fh> };
    close($fh);
    my $config = json_decode($content);
    $config->{src_connection} = $self->_getConnectionConfigByName($config->{src_connection_name});
    $config->{dst_connection} = $self->_getConnectionConfigByName($config->{dst_connection_name});
    return $config;
}

sub _getConnectionConfigByName
{
    my $self = shift;
    my $connection_name = shift;
    my $config_path = $Consts::CONFIG_PATH. $Consts::DS. 'connections'. $Consts::DS. "${connection_name}.json";
    unless (-f $config_path) {
        Log::error("Connection config [${connection_name}] not found.");
        return;
    }
    open(my $fh, '<', $config_path) or croak "Failed to open file: $!";
    my $content = do { local $/; <$fh> };
    close($fh);
    my $config = json_decode($content);
    return $config;
}

sub _getDB
{
    my $self = shift;
    my $config = shift;

    my $db_module_path = $Consts::LIB_PATH. $Consts::DS. 'DB'. $Consts::DS. $config->{type}. '.pm';
    my $db_module = "DB::" . $config->{type};
    require $db_module_path;
    return $db_module->new(\%$config);
}

sub _validateCount
{
    my ($self, $src_db, $dst_db, $table) = @_;

    my $src_count = $src_db->getCount($table);
    my $dst_count = $dst_db->getCount($table);
    Log::info("    src count: $src_count, dst count: $dst_count");
    return $src_count == $dst_count;
}

sub _validateColumnSum
{
    my ($self, $src_db, $dst_db, $table, $column) = @_;

    my $src_sum = $src_db->getColumnSum($table, $column);
    my $dst_sum = $dst_db->getColumnSum($table, $column);
    Log::info("    src sum: $src_sum, dst sum: $dst_sum");
    return $src_sum == $dst_sum;
}

sub _validateRows
{
    my ($self, $src_db, $dst_db, $table, $src_string_columns, $sampling_conditions, $order, $diff_row_limit) = @_;

    my $sql = sprintf($src_db->getSql('gs-table-ampling-rows'), '*', $src_db->{_config}->{schema}, $table). " ${sampling_conditions}";
    my @src_rows = $src_db->getSamplingRows($table, $sampling_conditions, $order, $diff_row_limit);
    my @dst_rows = $dst_db->getSamplingRows($table, $sampling_conditions, $order, $diff_row_limit);

    my $validate_rows_res = 1;
    my @rows;

    for my $i (0..$#src_rows) {
        my $src_row = $src_rows[$i];
        my $dst_row = $dst_rows[$i];

        my @row;
        for my $string_column (@$src_string_columns) {
            my $src_value = $src_row->{$string_column};
            my $dst_value = $dst_row->{$string_column};

            unless (defined $src_value) {
                $src_value = '';
            }
            unless (defined $dst_value) {
                $dst_value = '';
            }

            if ($src_value eq $dst_value) {
                Log::debug("    Row [${i}], column [ $string_column ] match. src: [ $src_value ], dst: [ $dst_value ]");
            } else {
                $validate_rows_res = 0;
                Log::warn("    Row [${i}], column [ $string_column ] mismatch! src: [ $src_value ], dst: [ $dst_value ]");
            }
            push @row, (
                $src_value,
                $dst_value,
                $src_value eq $dst_value ? 1 : 0
            );
        }

        push @rows, \@row;
    }

    return $validate_rows_res, @rows;
}

1;